package com.hk.stream.rocketmq.appender.log4j2;

import com.hk.commons.util.JsonUtils;
import com.hk.commons.util.SpringContextHolder;
import com.hk.commons.util.StringUtils;
import com.hk.stream.rocketmq.NullSendCallback;
import org.apache.rocketmq.client.impl.CommunicationMode;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;

import java.time.Instant;
import java.util.*;

/**
 * @author Kevin
 * @date 2023-05-27 23:08
 */
public class RocketMqSender extends TimerTask {

    private final String topic;

    private final String host;

    private final String source;

    private final String index;

    private final int maxEventsBatchCount;

    private final CommunicationMode communicationMode;

    private Timer timer;

    private RocketMQTemplate rocketMQTemplate;

    private List<String> eventsBatch = new LinkedList<>();

    public RocketMqSender(String topic, String host, String source, String index, int maxEventsBatchCount,
                          String communicationMode, long delay) {
        this.topic = topic;
        this.host = host;
        this.source = source;
        this.index = index;
        this.maxEventsBatchCount = maxEventsBatchCount <= 0 ? 30 : maxEventsBatchCount;
        this.communicationMode = StringUtils.hasLength(communicationMode) ?
                CommunicationMode.valueOf(communicationMode) : CommunicationMode.ONEWAY;
        if (delay > 0) { // 单位: 毫秒
            this.timer = new Timer();
            this.timer.scheduleAtFixedRate(this, delay, delay);
        }
    }

    public RocketMQTemplate getRocketMQTemplate() {
        if (Objects.isNull(this.rocketMQTemplate)) {
            this.rocketMQTemplate = SpringContextHolder.getBeanIfExist(RocketMQTemplate.class).orElse(null);
        }
        return rocketMQTemplate;
    }

    public synchronized void send(String severity, String message, String loggerName, String threadName, String exceptionMessage) {
        Map<String, String> map = HashMap.newHashMap(6);
        map.put("time", String.valueOf(Instant.now().toEpochMilli() / 1000));
        map.put("severity", severity);
        map.put("message", message);
        map.put("logger_name", loggerName);
        map.put("thread_name", threadName);
        if (StringUtils.hasLength(exceptionMessage)) {
            map.put("exception_message", exceptionMessage);
        }
        eventsBatch.add(JsonUtils.serialize(map));
        if (eventsBatch.size() >= maxEventsBatchCount) {
            flush();
        }
    }

    private synchronized void flush() {
        if (!this.eventsBatch.isEmpty()) {
            RocketMQTemplate template = getRocketMQTemplate();
            if (Objects.nonNull(template)) {
                Message<List<String>> message = MessageBuilder.withPayload(this.eventsBatch)
                        .setHeader("host", host)
                        .setHeader("source", source)
                        .setHeader("index", index)
                        .build();
                try {
                    switch (communicationMode) {
                        // 同步发送消息
                        case SYNC -> template.syncSend(this.topic, message);
                        // 异步发送消息
                        case ASYNC -> template.asyncSend(this.topic, message, NullSendCallback.INSTANCE);
                        //  直接发送
                        // 采用one-way发送模式发送消息的时候，发送端发送完消息后会立即返回，不会等待来自broker的ack来告知本次消息发送是否完全完成发送。
                        // 这种方式吞吐量很大，但是存在消息丢失的风险，所以其适用于不重要的消息发送，比如日志收集。
                        case ONEWAY -> template.sendOneWay(this.topic, message);
                    }
                } catch (Exception e) {
//                System.err.println("fail to Send Message:" + message.getPayload());
                }
            }

            this.eventsBatch = new LinkedList<>();
        }
    }


    @Override
    public void run() {
        flush();
    }

    public void close() {
        if (Objects.nonNull(this.timer)) {
            this.timer.cancel();
        }
        flush();
        super.cancel();
    }
}
